package com.codeduck.service.consumer;

import com.codeduck.kafka.KafkaInvoker;
import com.codeduck.kafka.consumer.ConsumerProperties;
import com.codeduck.pojo.TopicHandler;
import com.codeduck.utils.NetworkUtils;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.Properties;

/**
 * @auther Mahone Yan
 * @date 2023/7/16
 */
@Component
public class SystemLogConsumer extends AbstractConsumerService {

    private static final Logger LOGGER = LoggerFactory.getLogger(AcctDetailConsumer.class);
    private static final String GROUP_ID = "APP_SYSTEM_LOG_CONSUMER_GROUP";
    private static final String CLIENT_ID = "CID_SYSTEM_LOG_CONSUMER_" + NetworkUtils.getIP();

    @Resource
    private KafkaInvoker kafkaInvoker;

    @PostConstruct
    public void init() {
        this.start();
    }

    @Override
    void start() {
        Properties properties = ConsumerProperties.getProperties();
        properties.put("group.id", GROUP_ID);
        properties.put(AdminClientConfig.CLIENT_ID_CONFIG, CLIENT_ID);
        try {
            this.kafkaInvoker.subscribe(properties, TopicHandler.SYSTEM_LOG_TOPIC);
            LOGGER.info("================> Kafka 系统日志消费者线程已启动完毕 <================");
        } catch (Exception e) {
            LOGGER.error("系统日志消费者线程启动失败", e);
        }
    }
}